package com.jcloud.codetest1.component;

import com.jcloud.codetest1.domain.RecordItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;


/**
 * 功能描述：传输记录去消费
 *
 * @param:
 * @Return:
 * @Author:  2021/4/18 11:18
 */
@Component
public class RecordTransport implements InitializingBean {
    private static final Logger LOG = LoggerFactory.getLogger(RecordComsumer.class);

    @Autowired
    RecordComsumer recordComsumer;

    @Autowired
    RecordTask recordTask;

    @Autowired
    RecordItemBlockingQueue recordItemBlockingQueue;

    @Autowired
    FailRecordItemBlockingQueue failRecordItemBlockingQueue;

    @Resource(name = "recordExecutor")
    ThreadPoolTaskExecutor recordExecutor;

    @Resource(name = "failRecordExecutor")
    ThreadPoolTaskExecutor failRecordExecutor;

    /**
     * 功能描述：生产者
     *
     * @param: []
     * @Return: void
     * @Author:  2021/4/18 14:24
     */
    public void producer() {
        //记录处理任务的偏移量
        int offset = 0;
        RecordItem recordItem = null;
        while (true) {
            while(true) {
                //避免队列塞满
                if (recordItemBlockingQueue.remainingCapacity()== 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    break;
                }
            }
            try {
                recordItem = RecordTask.recordItemList.get(offset);
            } catch (IndexOutOfBoundsException e1) {
                continue;
            }
            recordItemBlockingQueue.putQueue(recordItem);
            offset++;
        }
    }

    /**
     * 功能描述： 消费正常队列
     *
     * @param: []
     * @Return: void
     * @Author:  2021/4/18 13:55
     */
    public void consumer() {
        while (true) {
            while(true) {
                //避免队列塞满
                if ( recordExecutor.getThreadPoolExecutor().getQueue().remainingCapacity() == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    break;
                }
            }
            //获取数据进行消费
            RecordItem recordItem = recordItemBlockingQueue.takeQueue();
            try {
                //LOG.info("取出记录{}", recordItem.getGenTime());
                recordExecutor.submit(() -> {
                    try {
                        consumerItem(recordItem);
                    } catch (RuntimeException e){
                        failRecordItemBlockingQueue.putQueue(recordItem);
                    } catch (Exception e1) {
                        e1.printStackTrace();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
                LOG.error("添加到消费线程池失败 {}", e.getMessage());
            }
        }
    }

    /**
     * 功能描述：消费失败队列
     *
     * @param: []
     * @Return: void
     * @Author:  2021/4/18 14:22
     */
    public void consumerFail() {
        while (true) {
            while(true) {
                //避免队列塞满
                if ( failRecordExecutor.getThreadPoolExecutor().getQueue().remainingCapacity() == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    break;
                }
            }
            //获取数据进行消费
            RecordItem recordItem = failRecordItemBlockingQueue.takeQueue();
            try {
                //LOG.info("取出记录{}", recordItem.getGenTime());
                failRecordExecutor.submit(() -> {
                    try {
                        consumerItem(recordItem);
                    } catch (RuntimeException e){
                        failRecordItemBlockingQueue.putQueue(recordItem);
                    } catch (Exception e1) {
                        e1.printStackTrace();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
                LOG.error("添加到消费线程池失败 {}", e.getMessage());
            }
        }
    }


    /**
     * 功能描述：消费
     *
     * @param: [recordItem]
     * @Return: void
     * @Author:  2021/4/18 14:24
     */
    private void consumerItem(RecordItem recordItem) throws InterruptedException {
        RecordItem result = recordComsumer.comsumer(recordItem);
        //LOG.info("消费结果{} {}", result.getGenTime(), result.getStatus());
        if (result.getStatus().equalsIgnoreCase("U")) {
            throw new RuntimeException("消费者处理结果位置");
        }
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        new Thread(() -> producer()).start();
        new Thread(() -> consumer()).start();
        new Thread(() -> consumerFail()).start();
    }


}
